package com.jwater.core.manager.proxy;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

import com.jwater.core.manager.common.MsgKeys;
import com.jwater.core.manager.common.model.FileInfo;
import com.jwater.core.manager.master.event.AlarmReportInfoEvent;
import com.jwater.core.manager.master.event.ProxyReportInfoEvent;
import jwater.org.apache.commons.io.FileUtils;
import jwater.org.apache.mina.core.IoUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.jwater.core.common.S;
import com.jwater.core.manager.common.DataKeys;
import com.jwater.core.manager.common.ManagerException;
import com.jwater.core.manager.common.model.JobInfo;
import com.jwater.core.manager.common.model.JvmInfo;
import com.jwater.core.manager.common.model.NodeInfo;
import com.jwater.core.manager.master.MasterFactory;
import com.jwater.core.manager.master.event.CompleteJobEvent;
import com.jwater.core.manager.proxy.cmd.CmdClient;
import com.jwater.core.manager.proxy.cmd.CmdException;
import com.jwater.core.manager.proxy.cmd.CmdResult;
import com.jwater.core.net.FileHandler;
import com.jwater.core.net.Node;
import com.jwater.core.net.NodeHandler;
import com.jwater.core.net.SimpleNode;
import com.jwater.core.serialize.SerializeUtils;
import com.jwater.launch.HomePath;

/**
 * @description: proxy客户端
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class ProxyClient implements CmdClient, NodeHandler, FileHandler {
    protected static Logger logger = LoggerFactory.getLogger(ProxyClient.class);

    private Node clientNode;
    private Map<String, CountDownLatch> getFileLatchs = new ConcurrentHashMap<>();
    private Map<String, String> fileErrorMsgs = new ConcurrentHashMap<>();
    private Map<String, PrintStream> outFiles = new ConcurrentHashMap<>();
    private NodeInfo nodeInfo;
    private int masterId = -1;
    private long uptime;

    public void setMasterId(int masterId) {
        this.masterId = masterId;
    }

    /**
     * 启动
     *
     * @param host
     * @param port
     * @param nodeInfo
     */
    public void start(String host, int port, NodeInfo nodeInfo) {
        this.nodeInfo = nodeInfo;
        if (nodeInfo == null) {
            throw new ManagerException("can not find node info for:" + host);
        }
        clientNode = new SimpleNode();
        clientNode.setHandler(this);
        clientNode.setFileHandler(this);
        clientNode.startAsClient(host, port, true);
    }

    /**
     * 停止
     */
    public void stop() {
        if (clientNode != null) {
            clientNode.close();
        }
    }

    /**
     * 获取上线时间
     *
     * @return
     */
    public long getUptime() {
        return uptime;
    }

    /**
     * 主机信息
     *
     * @return
     */
    public NodeInfo getNodeInfo() {
        return nodeInfo;
    }

    /**
     * 是否连着的
     *
     * @return
     */
    public boolean isConnect() {
        return clientNode.isConnect();
    }

    /**
     * 获取建立连接的次数
     *
     * @return
     */
    public int getConnectCount() {
        return clientNode.getConnectCount();
    }

    /**
     * 转化结构对象
     *
     * @param bytes
     * @return
     */
    private CmdResult toCmdResult(byte[] bytes) {
        CmdResult result = new CmdResult();
        if (bytes.length > 0) {
            result = result.deserialize(bytes);
        } else {
            result.setSuccess(false);
            throw new CmdException("net error");
        }
        if (!result.getSuccess()) {
            throw new CmdException(result.getMsg());
        }
        return result;
    }


    @Override
    public void startMaster(int id) {
        byte[] bytes = clientNode.syncCall("startMaster", S.toBytes(id + ""));
        toCmdResult(bytes);
    }

    @Override
    public void stopMaster(int id) {
        byte[] bytes = clientNode.syncCall("stopMaster", S.toBytes(id + ""));
        toCmdResult(bytes);
    }

    @Override
    public void startProxy() {
        byte[] bytes = clientNode.syncCall("startProxy", new byte[0]);
        toCmdResult(bytes);
    }

    @Override
    public void stopProxy() {
        byte[] bytes = clientNode.syncCall("stopProxy", new byte[0]);
        toCmdResult(bytes);
    }

    @Override
    public void startWorker(String host, String regionName, String deployName, String groupName, String workerName,
                            int index, String jvmArgs) {
        byte[] bytes = SerializeUtils.serializeArray(host, regionName, deployName, groupName, workerName, index,
                jvmArgs);
        bytes = clientNode.syncCall("startWorker", bytes);
        toCmdResult(bytes);
    }

    @Override
    public void stopWorker(String regionName, String deployName, String groupName, String workerName, int index,
                           boolean isForce) {
        byte[] bytes = SerializeUtils.serializeArray(regionName, deployName, groupName, workerName, index, isForce);
        bytes = clientNode.syncCall("stopWorker", bytes);
        toCmdResult(bytes);
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<String> cmd(String cmd) {
        byte[] bytes = clientNode.syncCall("cmd", S.toBytes(cmd));
        return (List<String>) toCmdResult(bytes).getData();
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<String> listFiles(String dstDir, boolean recursive) {
        byte[] bytes = SerializeUtils.serializeArray(dstDir, recursive);
        bytes = clientNode.syncCall("listFiles", bytes);
        CmdResult result = toCmdResult(bytes);
        if (result.getData() != null) {
            return (List<String>) result.getData();
        } else {
            return new ArrayList<String>();
        }
    }

    @Override
    public void putFile(InputStream src, String dst) {
        clientNode.sendFile(UUID.randomUUID().toString(), dst, src);
    }

    @Override
    public File getFile(String dst) {
        byte[] bytes = clientNode.syncCall("preGetFile", S.toBytes(dst));
        toCmdResult(bytes);
        String category = UUID.randomUUID().toString();
        String fileName = category;
        bytes = SerializeUtils.serializeArray(dst, category, fileName);
        CountDownLatch latch = new CountDownLatch(1);
        getFileLatchs.put(category, latch);
        clientNode.sendData("getFile", bytes);
        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.error(e.toString(), e);
        }
        File file = new File(HomePath.getTmpFilePath(), category);
        if (file.exists()) {
            return file;
        }
        // 错误处理
        String errorMsg = fileErrorMsgs.remove(category);
        if (errorMsg != null) {
            throw new CmdException(errorMsg);
        }
        return new File(HomePath.getTmpFilePath(), category);
    }

    @Override
    public void removeFile(String dst) {
        byte[] bytes = clientNode.syncCall("removeFile", S.toBytes(dst));
        toCmdResult(bytes);
    }

    @Override
    public String getContent(String dst, int lineLimit, int lineLenLimit) {
        byte[] bytes = SerializeUtils.serializeArray(dst, lineLimit, lineLenLimit);
        bytes = clientNode.syncCall("getContent", bytes);
        CmdResult result = toCmdResult(bytes);
        return (String) result.getData();
    }

    @Override
    public void saveContent(String dst, String content) {
        byte[] bytes = SerializeUtils.serializeArray(dst, content);
        bytes = clientNode.syncCall("saveContent", bytes);
        toCmdResult(bytes);
    }

    @Override
    public void deploy(String fileName, String dst, String regionName, String deployName, String keepConf, long version,
                       String path) {
        byte[] bytes = SerializeUtils.serializeArray(fileName, dst, regionName, deployName, keepConf, version, path);
        bytes = clientNode.syncCall("deploy", bytes);
        toCmdResult(bytes);
    }

    @Override
    public void removeDeploy(String regionName, String deployName) {
        byte[] bytes = SerializeUtils.serializeArray(regionName, deployName);
        bytes = clientNode.syncCall("removeDeploy", bytes);
        toCmdResult(bytes);
    }

    @Override
    public void runJob(String regionName, String deployName, String groupName, String jobName, long time,
                       boolean once) {
        byte[] bytes = SerializeUtils.serializeArray(regionName, deployName, groupName, jobName, time, once);
        bytes = clientNode.syncCall("runJob", bytes);
        toCmdResult(bytes);
    }

    @Override
    public void killJob(String regionName, String deployName, String groupName, String jobName) {
        byte[] bytes = SerializeUtils.serializeArray(regionName, deployName, groupName, jobName);
        bytes = clientNode.syncCall("killJob", bytes);
        toCmdResult(bytes);
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<String> listLogFiles(String regionName, String dir) {
        byte[] bytes = SerializeUtils.serializeArray(regionName, dir);
        bytes = clientNode.syncCall("listLogFiles", bytes);
        return (List<String>) toCmdResult(bytes).getData();
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<JvmInfo> listJvmInfo() {
        byte[] bytes = clientNode.syncCall("listJvmInfo", S.toBytes(""));
        return (List<JvmInfo>) toCmdResult(bytes).getData();
    }

    @Override
    public File cmdJvmFile(String cmdType, int pid) {
        byte[] bytes = clientNode.syncCall("cmdJvm", SerializeUtils.serializeArray(cmdType, pid));
        String dst = (String) toCmdResult(bytes).getData();

        String category = UUID.randomUUID().toString();
        String fileName = category;
        bytes = SerializeUtils.serializeArray(dst, category, fileName);
        CountDownLatch latch = new CountDownLatch(1);
        getFileLatchs.put(category, latch);
        clientNode.sendData("cmdJvmFile", bytes);
        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.error(e.toString(), e);
        }
        return new File(HomePath.getTmpFilePath(), category);
    }

    @Override
    public List<FileInfo> listFileInfos(String dstDir) {
        byte[] bytes = clientNode.syncCall("listFileInfos", S.toBytes(dstDir));
        return (List<FileInfo>) toCmdResult(bytes).getData();
    }

    @Override
    public void changeDeployProp(String regionName, String deployName, String key, String value) {
        byte[] bytes = SerializeUtils.serializeArray(regionName, deployName, key, value);
        bytes = clientNode.syncCall("changeDeployProp", bytes);
        toCmdResult(bytes).getData();
    }

    @Override
    public void onConnect(Node node) {
        uptime = System.currentTimeMillis();
        logger.info("proxy connect:" + node.getNodeId());
        if (masterId != -1) {
            clientNode.syncCall("login", S.toBytes(String.valueOf(masterId)));
        }
    }

    @Override
    public void onDisconnect(Node node) {
        logger.info("proxy disconnect:" + node.getNodeId());
    }

    @Override
    public void onData(Node node, String dataType, byte[] data) {
        if (MsgKeys.PROXY_REPORT.equals(dataType)) {
            ProxyReportInfoEvent event = new ProxyReportInfoEvent();
            event.setData(data);
            MasterFactory.getMasterContext().getMasterNode().dispatchEvent(event);
        } else if (MsgKeys.ALARM_REPORT.equals(dataType)) {
            AlarmReportInfoEvent event = new AlarmReportInfoEvent();
            event.setData(data);
            MasterFactory.getMasterContext().getMasterNode().dispatchEvent(event);
        }
    }

    @Override
    public byte[] onSyncCall(Node node, String method, byte[] args) {
        if ("completeJob".equals(method)) {
            doForCompleteJob(args);
        }
        return null;
    }

    private void doForCompleteJob(byte[] args) {
        logger.info("doForCompleteJob");
        Object[] objs = SerializeUtils.deserializeArray(args);
        String regionName = (String) objs[0];
        String deployName = (String) objs[1];
        String groupName = (String) objs[2];
        String jobName = (String) objs[3];
        long time = (long) objs[4];
        JobInfo jobInfo = MasterFactory.getMasterContext().getMasterNode().getJobConfParse().getJobInfo(regionName,
                deployName, groupName, jobName);
        if (jobInfo == null) {
            throw new ManagerException(
                    "can not find job for:" + DataKeys.getJobKey(regionName, deployName, groupName, jobName));
        }
        jobInfo.setTime(time);
        CompleteJobEvent event = new CompleteJobEvent();
        event.setJobInfo(jobInfo);
        MasterFactory.getMasterContext().getMasterNode().dispatchEvent(event);
    }

    @Override
    public void onAsyncCall(Node node, String method, byte[] args, ReturnCallback callback) {
    }

    @Override
    public void onBegin(String category, String fileName) {
        try {
            File dir = new File(HomePath.getTmpFilePath());
            if (!dir.exists()) {
                dir.mkdirs();
            }
            PrintStream ps = new PrintStream(new File(dir, category));
            outFiles.put(category, ps);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public void onData(String category, String fileName, int index, byte[] data) {
        try {
            PrintStream ps = outFiles.get(category);
            ps.write(data);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public void onEnd(String category, String fileName, int index) {
        try {
            PrintStream ps = outFiles.remove(category);
            ps.close();
            getFileLatchs.remove(category).countDown();
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public void onError(String category, String fileName, String errorMsg) {
        getFileLatchs.remove(category).countDown();
        fileErrorMsgs.put(category, errorMsg);
    }
}
